package com.reload.stt.spark.real_ime;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.reload.utils.DraftWithOrigin;
import com.reload.utils.EncryptUtil;
import com.reload.utils.FileUtils;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.springframework.web.multipart.MultipartFile;
import reactor.core.publisher.Flux;

import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.io.File;
import java.io.RandomAccessFile;
import java.net.URI;
import java.net.URLEncoder;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.Objects;

/**
 * @Date: 2024/12/6 - 20:02
 * @Version: 17
 * @Author: Reloading
 * @Description: 使用Java-websocket 依赖包
 */
@Slf4j
public class WebSocketRequestClient {
    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");
    private static final String APPID = "bfbf0b9b";
    private static final String SECRET_KEY = "df45a4ffba481f5ae57a55c34d6d7bdd";
    // 请求地址
    private static final String HOST = "rtasr.xfyun.cn/v1/ws";
    private static final String BASE_URL = "wss://" + HOST;
    private static final String ORIGIN = "https://" + HOST;
    // 每次发送的数据大小 1280 字节
    private static final int CHUNCKED_SIZE = 1280;


    public WebSocketRequestClient() {

    }

    private static String getCurrentTimeStr() {
        return sdf.format(new Date());
    }

    // 生成握手参数
    private static String getHandShakeParams(String appId, String secretKey) {
        String ts = System.currentTimeMillis() / 1000 + "";
        String signa = "";
        try {
            signa = EncryptUtil.HmacSHA1Encrypt(EncryptUtil.MD5(appId + ts), secretKey);
            return "?appid=" + appId + "&ts=" + ts + "&signa=" + URLEncoder.encode(signa, "UTF-8");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "";
    }


    public Flux<String> sendFile(MultipartFile file) {

        return Flux.create(sink -> {
            try {
                URI url = new URI(BASE_URL + getHandShakeParams(APPID, SECRET_KEY));
                DraftWithOrigin draft = new DraftWithOrigin(ORIGIN);
                // 创建 WebSocket 客户端
                WebSocketClient client = new WebSocketClient(url, draft) {
                    @Getter
                    boolean started = false;
                    @Override
                    public void onOpen(ServerHandshake handshake) {
                        log.info("WebSocket connection opened.{}", handshake.getHttpStatusMessage());
                    }

                    @Override
                    public void onMessage(String msg) {
                        JSONObject msgObj = JSON.parseObject(msg);
                        String action = msgObj.getString("action");
                        if (Objects.equals("started", action)) {
                            // 握手成功
                            log.info("{}\t握手成功！sid: {}", getCurrentTimeStr(), msgObj.getString("sid"));
                            started = true;
                        } else if (Objects.equals("result", action)) {
                            // 转写结果
                            String data = getContent(msgObj.getString("data"));
                            // 将服务端返回的消息推送到 Flux
                            sink.next(data);
                        } else if (Objects.equals("error", action)) {
                            // 连接发生错误
                            log.error("Error: {}", msg);
                            onClose(1000, "错误关闭", true);
                            sink.complete();
                        }
                    }

                    @Override
                    public void onClose(int code, String reason, boolean remote) {
                        log.error("WebSocket connection closed: {}", reason);
                        sink.complete(); // 完成 Flux
                    }

                    @Override
                    public void onError(Exception ex) {
                        log.error("WebSocket connection error: {}", ex.getMessage());
                        sink.error(ex); // 推送错误到 Flux
                    }
                };

                if (url.toString().contains("wss")) {
                    trustAllHosts(client);
                }
                client.connectBlocking(); // 阻塞直到连接成功
                if (client.isOpen() ) {
                    // 发送数据
                    sendData(client, file);
                }
                // 处理取消事件
                sink.onDispose(client::close);
            } catch (Exception e) {
                // 如果出现异常，则推送错误
                sink.error(e);
            }
        });
    }

    private void sendData(WebSocketClient webSocketClient, MultipartFile file) {
        // MultipartFile 转File
        File tempFile = FileUtils.toFile(file);
        log.info("时间：{},开始发送音频数据", sdf.format(new Date()));
        // 发送音频
        byte[] bytes = new byte[CHUNCKED_SIZE];
        try (RandomAccessFile raf = new RandomAccessFile(tempFile, "r")) {
            int len = -1;
            long lastTs = 0;
            while ((len = raf.read(bytes)) != -1) {
                if (len < CHUNCKED_SIZE) {
                    bytes = Arrays.copyOfRange(bytes, 0, len);
                    webSocketClient.send(bytes);
                    break;
                }
                long curTs = System.currentTimeMillis();
                if (lastTs == 0) {
                    lastTs = System.currentTimeMillis();
                } else {
                    long s = curTs - lastTs;
                    if (s < 40) {
                        System.out.println("error time interval: " + s + " ms");
                    }
                }
                webSocketClient.send(bytes);
                // 每隔40毫秒发送一次数据
                Thread.sleep(40);
            }

            // 发送结束标识
            webSocketClient.send("{\"end\": true}");
            log.info("时间：{},发送结束标识完成", getCurrentTimeStr());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 把转写结果解析为句子
    private static String getContent(String message) {
        StringBuffer resultBuilder = new StringBuffer();
        try {
            JSONObject messageObj = JSON.parseObject(message);
            JSONObject cn = messageObj.getJSONObject("cn");
            JSONObject st = cn.getJSONObject("st");
            JSONArray rtArr = st.getJSONArray("rt");
            for (int i = 0; i < rtArr.size(); i++) {
                JSONObject rtArrObj = rtArr.getJSONObject(i);
                JSONArray wsArr = rtArrObj.getJSONArray("ws");
                for (int j = 0; j < wsArr.size(); j++) {
                    JSONObject wsArrObj = wsArr.getJSONObject(j);
                    JSONArray cwArr = wsArrObj.getJSONArray("cw");
                    for (int k = 0; k < cwArr.size(); k++) {
                        JSONObject cwArrObj = cwArr.getJSONObject(k);
                        String wStr = cwArrObj.getString("w");
                        resultBuilder.append(wStr);
                    }
                }
            }
        } catch (Exception e) {
            return message;
        }

        return resultBuilder.toString();
    }


    private void trustAllHosts(WebSocketClient client) {
        TrustManager[] trustAllCerts = new TrustManager[]{new X509TrustManager() {
            @Override
            public X509Certificate[] getAcceptedIssuers() {
                return new X509Certificate[]{};
            }

            @Override
            public void checkClientTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
                // Auto-generated method stub

            }

            @Override
            public void checkServerTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
                // Auto-generated method stub

            }
        }};

        try {
            SSLContext sc = SSLContext.getInstance("TLS");
            sc.init(null, trustAllCerts, new java.security.SecureRandom());
            client.setSocket(sc.getSocketFactory().createSocket());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
